﻿
using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace WalkthroughTDF
{
    /// <summary>
    /// BufferBlock is a unbound queue
    /// chekanote: BufferBlock<T> will hand out an individual element to only one receiver.  
    /// If multiple targets are linked from the buffer, it will offer a message to each in turn, 
    /// allowing only one to consume each.  And if code manually receives from the buffer, 
    /// doing so will remove the received element from the collection
    /// </summary>
    static class TestBufferBlock
    {
        #region "----------------------- internal classes -----------------------"
        class WorkerBase
        {
            protected int m_id;
            protected BufferBlock<int> m_queue;

            public WorkerBase(int id, BufferBlock<int> queue)
            {
                m_id = id;
                m_queue = queue;
            }
        }

        sealed class Producer : WorkerBase
        {
            public Producer(int id, BufferBlock<int> queue)
                : base(id, queue)
            {
            }

            public void Start(int total, int interval)
            {
                for (int index = 0; index < total; ++index)
                {
                    Thread.Sleep(interval);

                    int value = m_id + index;
                    m_queue.Post(value);
                    Console.WriteLine("+Producer[{0}]: {1}", m_id, value);
                }// for 
            }
        }

        sealed class Consumer : WorkerBase
        {
            public Consumer(int id, BufferBlock<int> queue)
                : base(id, queue)
            {
            }

            public void Start()
            {
                m_queue.ReceiveAsync().ContinueWith(this.OnRecvedCallback);
            }

            private void OnRecvedCallback(Task<int> compTask)
            {
                // chekanote: there is no common shared variables existing between threads
                // accessing shared variables will not have race, because continue-thread will only run after the previous
                // thread, so there will never be two threads co-exists
                // but the serious problem is the "memory barier", which will prevent the changes happend
                // in one thread visbile to another thread
                // so I cannot use a integer to count the number having been received, unless I use lock or Interlocked
                Console.WriteLine("\t-Consumer[{0}]: {1}", m_id, compTask.Result);
                Start();
            }
        }
        #endregion

        public static void TestMain()
        {
            BufferBlock<int> queue = new BufferBlock<int>();

            Task.Factory.StartNew(() => new Producer(1000, queue).Start(10, 500));
            Task.Factory.StartNew(() => new Producer(2000, queue).Start(5, 1000));

            new Consumer(1, queue).Start();
            new Consumer(2, queue).Start();

            Console.WriteLine("Press any key to exit,......");
            Console.ReadLine();
        }
    }
}
